/**
 * Copyright (c) The openTCS Authors.
 * <p>
 * This program is free software and subject to the MIT license. (For details,
 * see the licensing information (LICENSE.txt) you should have received with
 * this copy of the software.)
 */
package org.opentcs.strategies.basic.scheduling;

import jakarta.annotation.Nonnull;

import static java.util.Objects.requireNonNull;

import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;

import org.opentcs.components.kernel.Scheduler;
import org.opentcs.components.kernel.Scheduler.Client;
import org.opentcs.customizations.kernel.GlobalSyncObject;
import org.opentcs.data.model.TCSResource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Handles regular resource allocations.
 */
class AllocatorTask
    implements Runnable {

  /**
   * This class's Logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger(AllocatorTask.class);
  /**
   * The reservation pool.
   */
  private final ReservationPool reservationPool;
  /**
   * Takes care of (sub)modules.
   */
  private final Scheduler.Module allocationAdvisor;
  /**
   * Allocations deferred because they couldn't be granted, yet.
   */
  private final Queue<AllocatorCommand.Allocate> deferredAllocations;
  /**
   * Executes tasks.
   */
  private final ScheduledExecutorService kernelExecutor;
  /**
   * A global object to be used for synchronization within the kernel.
   */
  private final Object globalSyncObject;
  /**
   * Describes the actual task.
   */
  private final AllocatorCommand command;

  private final Map<Client, Set<Future<?>>> scheduleRetryWaitingFutures;


  /**
   * Creates a new instance.
   */
  AllocatorTask(@Nonnull ReservationPool reservationPool,
                @Nonnull Queue<AllocatorCommand.Allocate> deferredAllocations,
                @Nonnull Scheduler.Module allocationAdvisor,
                @Nonnull ScheduledExecutorService kernelExecutor,
                @Nonnull @GlobalSyncObject Object globalSyncObject,
                @Nonnull AllocatorCommand command,
                Map<Client, Set<Future<?>>> scheduleRetryWaitingFutures) {
    this.reservationPool = requireNonNull(reservationPool, "reservationPool");
    this.deferredAllocations = requireNonNull(deferredAllocations, "deferredAllocations");
    this.allocationAdvisor = requireNonNull(allocationAdvisor, "allocationAdvisor");
    this.kernelExecutor = requireNonNull(kernelExecutor, "kernelExecutor");
    this.globalSyncObject = requireNonNull(globalSyncObject, "globalSyncObject");
    this.command = requireNonNull(command, "command");
    this.scheduleRetryWaitingFutures = requireNonNull(scheduleRetryWaitingFutures, "scheduleRetryWaitingFutures");
  }

  @Override
  public void run() {
    LOG.debug("Processing AllocatorCommand: {}", command);

    if (command instanceof AllocatorCommand.Allocate) {
      processAllocate((AllocatorCommand.Allocate) command);
    } else if (command instanceof AllocatorCommand.RetryAllocates) {
      scheduleRetryWaitingAllocations();
    } else if (command instanceof AllocatorCommand.CheckAllocationsPrepared) {
      checkAllocationsPrepared((AllocatorCommand.CheckAllocationsPrepared) command);
    } else if (command instanceof AllocatorCommand.AllocationsReleased) {
      allocationsReleased((AllocatorCommand.AllocationsReleased) command);
    } else {
      LOG.warn("Unhandled AllocatorCommand implementation {}, ignored.", command.getClass());
    }
  }

  private void processAllocate(AllocatorCommand.Allocate command) {
    if (!tryAllocate(command)) {
      LOG.debug("{}: Resources unavailable, deferring allocation...", command.getClient().getId());

      return;
    }
    checkAllocationsPrepared(command.getClient(), command.getResources());
  }

  private void checkAllocationsPrepared(AllocatorCommand.CheckAllocationsPrepared command) {
    checkAllocationsPrepared(command.getClient(), command.getResources());
  }

  private void checkAllocationsPrepared(Client client, Set<TCSResource<?>> resources) {
    if (!allocationAdvisor.hasPreparedAllocation(client, resources)) {
      LOG.debug("{}: Preparation of resources not yet done.",
          client.getId());
      // XXX remember the resources a client is waiting for preparation done?
      return;
    }

    LOG.debug("Preparation of resources '{}' successful, calling back client '{}'...",
        resources,
        client.getId());
    if (!client.allocationSuccessful(resources)) {
      LOG.warn("{}: Client didn't want allocated resources ({}), unallocating them...",
          client.getId(),
          resources);
      undoAllocate(client, resources);
      // See if others want the resources this one didn't, then.
      scheduleRetryWaitingAllocations();
    }
    // Notify modules about the changes in claimed/allocated resources for this client.
    allocationAdvisor.setAllocationState(client,
        reservationPool.allocatedResources(client),
        reservationPool.getClaim(client));
  }

  /**
   * Allocates the given set of resources, if possible.
   *
   * @param command Describes the requested allocation.
   * @return <code>true</code> if, and only if, the given resources were allocated.
   */
  private boolean tryAllocate(AllocatorCommand.Allocate command) {
    Scheduler.Client client = command.getClient();
    Set<TCSResource<?>> resources = command.getResources();

    synchronized (globalSyncObject) {
      if (!reservationPool.isNextInClaim(client, resources)) {
        LOG.error("{}: Not allocating resources that are not next claimed resources: {}",
            client.getId(),
            resources);
        List<Set<TCSResource<?>>> claim = reservationPool.getClaim(command.getClient());
        if (claim.contains(resources)) {
          deferredAllocations.add(command);
        } else {
          //如果是重路由失败了那么就直接进行一次尝试
//          scheduleRetryWaitingAllocations();
        }
        return false;
      }

      LOG.debug("{}: Checking resource availability: {}...", client.getId(), resources);
      if (!reservationPool.resourcesAvailableForUserWithBlock(resources, client)) {
        LOG.debug("{}: Resources unavailable.", client.getId());
        //如果造成了环并且是自己进行了重路由，那么就直接不添加重试----(有问题)//todo:这里需要解决一下为啥不存在的问题
        scheduleRetryWaitingFutures.putIfAbsent(client, new HashSet<>());
        reservationPool.pushBlockEvent(client.getId());
        //如果是自己进行了重路由那么就不需要添加到重试队列当中。(如果不存在client这个key了那么说明该client进行了重路由操作。)
        if (reservationPool.isCaseRing(client)) {
          reservationPool.pushDyEvent();
          if (scheduleRetryWaitingFutures.get(client) == null) {
            LOG.info("{}申请{}自身发生重路由，不进行资源重申", client.getId(), Arrays.toString(command.getResources().toArray()));
            return false;
          }
        }
        deferredAllocations.add(command);
        LOG.info("{}添加{}资源到AllocateTask重试队列中", client.getId(), command.getResources().stream().findFirst().get().getName());
        return false;
      }

      LOG.debug("{}: Checking if resources may be allocated...", client.getId());
      HashSet<TCSResource<?>> tcsResources = new HashSet<>(resources);
      if (command.getRelayResources() != null)
        tcsResources.add(command.getRelayResources());
      if (!allocationAdvisor.mayAllocate(client, tcsResources)) {
        LOG.debug("{}: Resources may not be allocated.", client.getId());
        deferredAllocations.add(command);
        return false;
      }

      LOG.debug("{}: Preparing resources for allocation...", client.getId());
      allocationAdvisor.prepareAllocation(client, resources);


      LOG.debug("{}: All resources available, allocating...", client.getId());
      // Allocate resources.
      for (TCSResource<?> curRes : resources) {
        reservationPool.getReservationEntry(curRes).allocate(client);
      }

      LOG.debug("{}: Removing resources claim: {}...", client.getId(), resources);
      reservationPool.unclaim(client, resources);
      //分配成功之后，需要更新前端展示数据
      reservationPool.pushBlockEvent(client.getId());
      return true;
    }
  }

  private void allocationsReleased(AllocatorCommand.AllocationsReleased command) {
    allocationAdvisor.allocationReleased(command.getClient(), command.getResources());
  }

  /**
   * Unallocates the given set of resources.
   * <p>
   * Note that this does <em>not</em> return any previously claimed resources to the client!
   * </p>
   */
  private void undoAllocate(Client client, Set<TCSResource<?>> resources) {
    synchronized (globalSyncObject) {
      reservationPool.free(client, resources);
    }
  }

  /**
   * Moves all waiting allocations back into the incoming queue so they can be rechecked.
   */
  private void scheduleRetryWaitingAllocations() {
    for (AllocatorCommand.Allocate allocate : deferredAllocations) {
      Future<?> submit = kernelExecutor.submit(new AllocatorTask(reservationPool,
          deferredAllocations,
          allocationAdvisor,
          kernelExecutor,
          globalSyncObject,
          allocate, scheduleRetryWaitingFutures));
      Set<Future<?>> futures = scheduleRetryWaitingFutures.computeIfAbsent(allocate.getClient(), k -> new HashSet<>());
      futures.add(submit);
    }
    deferredAllocations.clear();
  }
}
